package com.weibo.dip.hatest;

import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;

import java.io.IOException;
import java.util.concurrent.TimeUnit;

/** Created by haisen on 2018/8/2. */
public class ZkExecutor implements Watcher {
  private static final String CONNECT_ADDR = "10.13.4.44:2181";
  private static final String PARENT_PATH = "/testHA";
  private static final int SESSION_TIMEOUT = 10000;

  private static ZooKeeper zooKeeper;

  /** initial connection. */
  public ZkExecutor() {
    connectServer();
  }

  private void connectServer() {
    try {
      zooKeeper = new ZooKeeper(CONNECT_ADDR, SESSION_TIMEOUT, this);
      System.out.println("Create zk successs");
    } catch (IOException e) {
      System.out.println("Create zk error");
    }
  }

  /** create node test. */
  public boolean createNode() {
    if (zooKeeper == null) {
      return false;
    }
    boolean bool = false;
    try {
      zooKeeper.create(PARENT_PATH, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
      addWatcher();
      System.out.println("Create znode success  Ready to start the jar ");
      bool = true;
      ProgramControl.start();
    } catch (KeeperException e) {
      bool = getKeeperException(e);
    } catch (InterruptedException e) {
      System.out.println(ExceptionUtils.getFullStackTrace(e));
    }
    return bool;
  }

  @Override
  public void process(WatchedEvent watchedEvent) {
    switch (watchedEvent.getType()) {
      case None:
        switch (watchedEvent.getState()) {
          case Expired:
            // The serving cluster has expired this session. 重连 可能会一直重连
            ProgramControl.stop();
            connectServer();
            addWatcher();
            System.out.println(
                "The serving cluster has expired a session ,"
                    + " connection again to get a new seeeion of ");
            break;
          case Disconnected:
            ProgramControl.stop();
            System.out.println("connection broken ");
            // todo send a watcher
            break;
          case SyncConnected:
            System.out.print("SyncConnected  ");
            addWatcher();
            break;
          default:
            System.out.print("AuthFailed / ConnectedReadOnly / SaslAuthenticated  ");
            addWatcher();
            // include AuthFailed / ConnectedReadOnly / SaslAuthenticated
            // todo send a watch
            break;
        }
        break;
      case NodeDeleted:
        System.out.println("Node was deleted  Ready to stop");
        ProgramControl.stop();
        while (!createNode()) {
          System.out.println("Create znode again ");
        }
        break;
      case NodeCreated:
        System.out.print("Create Node happened  ");
        addWatcher();
        break;
      default:
        System.out.print("Other watchedEvent.getType  ");
        addWatcher();
        break;
    }
  }

  /** remove znode , for test ,it's cause NodeDeleted. */
  public void removeZnode() {
    try {
      zooKeeper.delete(PARENT_PATH, -1);
    } catch (InterruptedException | KeeperException e) {
      e.printStackTrace();
    }
  }

  /** remove znode , for test ,it's cause NodeDeleted. */
  public void close() throws InterruptedException {
    zooKeeper.close();
  }

  private void addWatcher() {
    if (zooKeeper != null) {
      try {
        zooKeeper.exists(PARENT_PATH, true);
        System.out.println("add watch success");
      } catch (KeeperException e) {
        getKeeperException(e);
      } catch (InterruptedException e) {
        e.printStackTrace();
      }
    }
  }

  private boolean getKeeperException(KeeperException e) {
    boolean bool = false;
    System.out.println(e.code());
    switch (e.code()) {
      case NODEEXISTS:
        ProgramControl.stop();
        addWatcher();
        bool = true;
        break;
      case SESSIONEXPIRED:
        connectServer();
        break;
      case SESSIONMOVED:
        connectServer();
        break;
      case OK:
        bool = true;
        addWatcher();
        break;
      default:
        System.out.println("Other code");
        addWatcher();
        break;
    }
    return bool;
  }

  /** main test. */
  public static void main(String[] args) {

    Thread thread =
        new Thread(
            () -> {
              ZkExecutor zkExecutor = new ZkExecutor();
              zkExecutor.createNode();
              while (!Thread.interrupted()) {
                try {
                  TimeUnit.MILLISECONDS.sleep(15000);
                } catch (InterruptedException e) {
                  e.printStackTrace();
                }
                System.out.println("sleep 1 end");

                zkExecutor.removeZnode();

                try {
                  TimeUnit.MILLISECONDS.sleep(15000);
                } catch (InterruptedException e) {
                  e.printStackTrace();
                }
                System.out.println("sleep 2 end");
                // Thread.currentThread().interrupt();
              }
            });
    thread.start();
  }
}
